// Copyright (c) Microsoft Corporation. All rights reserved.

using System;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Diagnostics;
using System.Threading;

using Microsoft.TeamFoundation.Migration.Toolkit.Linking;

namespace Microsoft.TeamFoundation.Migration.Toolkit.Wit
{
    using ReadQueue = ProcessingQueue<IMigrationWorkItem>;
    using WriteQueue = ProcessingQueue<InternalChangeList>;
    using LinkOperation = SyncActionContainer<ILink>;

    /// <summary>
    /// Supported object.
    /// </summary>
    public enum SyncObjects
    {
        Artifacts,                                          // Artifacts will be processed
        Links,                                              // Links will be processed
    }

    /// <summary>
    /// Synchronization process.
    /// </summary>
    class SyncProcess : IDisposable
    {
        private SyncObjects m_syncObjects;                  // Object type
        private WorkItemTrackingSession m_session;          // Parent session
        private Thread m_thread;                            // Service thread
        private volatile System.Threading.ThreadState m_state;  // Thread's state
        private ManualResetEvent m_unlockEvent;             // Unlocked event
        private ManualResetEvent m_abortEvent;              // Process aborted event
        private ManualResetEvent m_completeEvent;           // Thread complete event
        private WaitHandle[] m_endEvents;                   // Events that end processing

        private object m_storesLock;                        // Synchronization object for work item stores
        private object m_sharedLock;                        // Shared synchronization object
        private Pair<IMigrationWorkItemStore> m_stores;     // Stores being synchronized
        private Pair<StorePool> m_storePools;               // Store pools
        private Pair<SystemType> m_storeTypes;              // Type of stores being synchronized
        private Side m_tfsSide;                             // Side where TFS system resides
        private FieldValueComparer m_tfsValueComparer;      // Comparer for field values on TFS side
        private ILinkEngine m_linkEngine;                   // Linking engine
        private Stats m_stats;                              // Statistics

        private ReadQueue m_slowQueue;                      // Slow analysis queue
        private ReadQueue m_fastQueue;                      // Fast analysis queue
        private Pair<WriteQueue> m_writeQueues;             // Write queues

        // Test events
        private ManualResetEvent m_readyEvent;              // Service is ready
        private ManualResetEvent m_readingEvent;            // Service is reading
        private ManualResetEvent m_writingEvent;            // Service is writing


        /// <summary>
        /// Returns parent session for the process.
        /// </summary>
        public WorkItemTrackingSession Session { get { return m_session; } }

        /// <summary>
        /// Returns mapper.
        /// </summary>
        public Mapper Mapper { get { return Session.Mapper; } }

        /// <summary>
        /// Returns linking engine.
        /// </summary>
        public ILinkEngine LinkEngine { get { return m_linkEngine; } }

        /// <summary>
        /// Returns synchronized systems.
        /// </summary>
        public Pair<IMigrationWorkItemStore> Stores 
        { 
            get 
            {
                if (m_stores == null)
                {
                    lock (m_storesLock)
                    {
                        if (m_stores == null)
                        {
                            var stores = new Pair<IMigrationWorkItemStore>(
                                Session.Stores[StoreTypes.Left],
                                Session.Stores[StoreTypes.Right]);

                            Side tfsSide = m_storeTypes[Side.Left] == SystemType.Tfs? Side.Left : Side.Right;

                            stores[!tfsSide].VerifySettings(
                                Session,
                                ((TfsMigrationWorkItemStore)stores[tfsSide]).WorkItemStore);
                            m_stores = stores;
                        }
                    }
                }
                return m_stores; 
            } 
        }

        /// <summary>
        /// Returns store pools.
        /// </summary>
        public Pair<StorePool> StorePools 
        { 
            get 
            {
                if (m_storePools == null)
                {
                    lock (m_sharedLock)
                    {
                        if (m_storePools == null)
                        {
                            m_storePools = new Pair<StorePool>(
                                new StorePool(Stores.Left), 
                                new StorePool(Stores.Right));
                        }
                    }
                }
                return m_storePools; 
            } 
        }

        /// <summary>
        /// Returns types of synchronized systems.
        /// </summary>
        public Pair<SystemType> StoreTypes { get { return m_storeTypes; } }

        /// <summary>
        /// Returns side where TFS store resides.
        /// </summary>
        public Side TfsSide { get { return m_tfsSide; } }

        /// <summary>
        /// Returns value comparer used on the TFS side.
        /// </summary>
        public FieldValueComparer TfsValueComparer 
        { 
            get 
            {
                if (m_tfsValueComparer == null)
                {
                    lock (m_sharedLock)
                    {
                        if (m_tfsValueComparer == null)
                        {
                            m_tfsValueComparer = new FieldValueComparer(Stores[TfsSide].StringValueComparer);
                        }
                    }
                }
                return m_tfsValueComparer; 
            } 
        }

        /// <summary>
        /// Returns statistics.
        /// </summary>
        public Stats Stats { get { return m_stats; } }

        /// <summary>
        /// Returns combined flags from both stores.
        /// </summary>
        public MigrationWorkItemData CombinedFlags { get { return Stores.Left.Flags & Stores.Right.Flags; } }

        /// <summary>
        /// Tells whether either store is flat, so historical revisions won't be migrated.
        /// </summary>
        public bool IsFlat { get { return m_session.IgnoreRevisions || (CombinedFlags & MigrationWorkItemData.Revisions) == 0; } }

        /// <summary>
        /// Returns working thread.
        /// </summary>
        public Thread Thread { get { return m_thread; } }

        /// <summary>
        /// Returns ready event (the service thread is working)
        /// </summary>
        public ManualResetEvent ReadyEvent { get { return m_readyEvent; } }

        /// <summary>
        /// Returns reading event (the process began processing updated items)
        /// </summary>
        public ManualResetEvent ReadingEvent { get { return m_readingEvent; } }

        /// <summary>
        /// Returns writing event (the process began writing to the write queue)
        /// </summary>
        public ManualResetEvent WritingEvent { get { return m_writingEvent; } }

        /// <summary>
        /// Constructor.
        /// </summary>
        /// <param name="syncObjects">Objects that must be synchronized</param>
        /// <param name="session">Session that owns the process</param>
        /// <param name="source">Type of the system to be used as a source of updated items</param>
        public SyncProcess(
            SyncObjects syncObjects,
            WorkItemTrackingSession session,
            SystemType source)
        {
            m_storesLock = new object();
            m_sharedLock = new object();
            m_syncObjects = syncObjects;
            m_session = session;
            m_thread = new Thread(new ThreadStart(ThreadProc));
            m_state = System.Threading.ThreadState.Unstarted;
            m_unlockEvent = new ManualResetEvent(true);
            m_abortEvent = new ManualResetEvent(false);
            m_completeEvent = new ManualResetEvent(false);
            m_endEvents = new WaitHandle[] { m_abortEvent, m_completeEvent };

            m_storeTypes = new Pair<SystemType>(SystemType.Tfs, SystemType.Other);
            m_tfsSide = Side.Left;

            m_stats = new Stats();

            m_linkEngine = session.LinkEngine;

            // Initialize queues
            Pair<QueueConfiguration> qc = new Pair<QueueConfiguration>(
                session.TfsMigrationSource.WriteQueueConfig,
                session.OtherMigrationSource.Endpoint.WriteQueueConfig);
            ReadQueue.ProcessItems readCallback = new ProcessingQueue<IMigrationWorkItem>.ProcessItems(AnalysisQueueCallback);

            m_slowQueue = new ProcessingQueue<IMigrationWorkItem>(session.SlowQueueConfig, readCallback);
            m_fastQueue = new ProcessingQueue<IMigrationWorkItem>(session.FastQueueConfig, readCallback);
            if (source == SystemType.Other)
            {
                SwapSides();
                qc = !qc;
            }
            m_writeQueues = new Pair<ProcessingQueue<InternalChangeList>>(
                new WriteQueue(qc.Left, new WriteQueue.ProcessItems(LeftWriteQueueCallback)),
                new WriteQueue(qc.Right, new WriteQueue.ProcessItems(RightWriteQueueCallback)));

            m_readyEvent = new ManualResetEvent(false);
            m_readingEvent = new ManualResetEvent(false);
            m_writingEvent = new ManualResetEvent(false);
        }

        /// <summary>
        /// Does a single synchronization pass from left to right.
        /// </summary>
        public void SinglePass()
        {
            if (m_syncObjects == SyncObjects.Artifacts)
            {
                // Process updated work items.
                IMigrationWorkItemStore src = Stores.Left;
                string watermark = Mapper.HighWatermark[StoreTypes.Left];

                TraceManager.TraceInformation("Getting modified items from '{0}!{1}'", src.SystemName, src.StoreName);
                IEnumerable<IMigrationWorkItem> items = src.GetItems(ref watermark);
                TraceManager.TraceInformation("Received modified items from '{0}!{1}'", src.SystemName, src.StoreName);

                ItemsSyncPass(items);
                Mapper.SaveWatermark(StoreTypes.Left, watermark);
            }
            else
            {
                Debug.Assert(m_syncObjects == SyncObjects.Links, "Invalid object type!");

                // Process deferred links
                if (LinkEngine == null)
                {
                    throw new MigrationException(Resources.ErrorNoLinkingEngine);
                }

                LinksSyncPass();
            }
        }

        /// <summary>
        /// Does a double synchronization pass from left to right.
        /// </summary>
        public void DoublePass()
        {
            m_state = System.Threading.ThreadState.Running;

            // Perform a two-way pass
            SinglePass();
            SwapSides();
            SinglePass();

            m_state = System.Threading.ThreadState.Stopped;
        }


        /// <summary>
        /// Synchronizes given items on the left side.
        /// </summary>
        /// <param name="ids">Ids of items to synchronize with</param>
        public void Synchronize(
            string[] ids)
        {
            if (m_syncObjects == SyncObjects.Links)
            {
                throw new MigrationException(Resources.ErrorInLinkingMode);
            }

            List<IMigrationWorkItem> items = null;

            for (int i = 0; i < ids.Length; i++)
            {
                Session.FireMigrationStart(this, new FullId(Stores.Left, ids[i]));
            }

            GetWorkItemResult[] results = Stores.Left.GetItems(ids);
            items = new List<IMigrationWorkItem>(results.Length);

            foreach (GetWorkItemResult res in results)
            {
                if (res.WorkItem != null)
                {
                    items.Add(res.WorkItem);
                }
                else
                {
                    ReportException(res.Id, res.Exception);
                }
            }

            ItemsSyncPass(items);
        }

        #region Service Methods

        /// <summary>
        /// Starts the synchronization in a separate thread
        /// </summary>
        public void Start()
        {
            TraceManager.TraceInformation("Starting WIT service thread...");
            m_abortEvent.Reset();
            m_completeEvent.Reset();
            m_thread = new Thread(new ThreadStart(ThreadProc));
            m_thread.Start();
            m_readyEvent.WaitOne();
        }

        /// <summary>
        /// Completes the synchronization process.
        /// </summary>
        public void Complete()
        {
            Trace.TraceInformation("Stopping WIT service thread...");
            m_unlockEvent.Set();
            m_state = System.Threading.ThreadState.StopRequested;
            m_completeEvent.Set();
            m_thread.Join();
        }

        /// <summary>
        /// Aborts the synchronization process.
        /// </summary>
        public void Abort()
        {
            Trace.TraceInformation("Aborting WIT service thread...");
            m_unlockEvent.Set();
            m_state = System.Threading.ThreadState.AbortRequested;
            m_abortEvent.Set();
            m_thread.Join();

            Debug.Assert(m_thread.ThreadState == System.Threading.ThreadState.Stopped, "Invalid thread state!");
        }

        /// <summary>
        /// Pauses the synchronization process.
        /// </summary>
        public void Pause()
        {
            TraceManager.TraceInformation("Pausing WIT service thread...");
            m_state = System.Threading.ThreadState.Suspended;
            m_unlockEvent.Reset();
        }

        /// <summary>
        /// Resumes the synchronization process after it was paused.
        /// </summary>
        public void Resume()
        {
            TraceManager.TraceInformation("Resuming WIT service thread...");
            m_state = System.Threading.ThreadState.Running;
            m_unlockEvent.Set();
        }

        #endregion

        #region Private Methods

        /// <summary>
        /// Service thread procedure.
        /// </summary>
        private void ThreadProc()
        {
            TraceManager.TraceInformation("WIT service thread has started!");
            m_state = System.Threading.ThreadState.Running;
            m_readyEvent.Set();
            Thread.CurrentThread.Name = "WIT Synchronization Thread";

            while (m_state != System.Threading.ThreadState.AbortRequested
                && m_state != System.Threading.ThreadState.StopRequested)
            {
                SinglePass();
                SwapSides();
                m_readingEvent.Reset();
                m_writingEvent.Reset();

                WaitHandle.WaitAny(m_endEvents, m_session.SleepTime * 1000, false);
            }

            // Wait for read queues
            if (m_state == System.Threading.ThreadState.StopRequested)
            {
                m_fastQueue.Flush();
                m_slowQueue.Flush();
            }
            WaitHandle.WaitAll(
                new WaitHandle[] { m_fastQueue.DoneEvent, m_slowQueue.DoneEvent });

            // Wait for write queues
            if (m_state == System.Threading.ThreadState.StopRequested)
            {
                m_writeQueues.Left.Flush();
                m_writeQueues.Right.Flush();
            }
            WaitHandle.WaitAll(
                new WaitHandle[] { m_writeQueues.Left.DoneEvent, m_writeQueues.Right.DoneEvent });

            if (m_state == System.Threading.ThreadState.StopRequested)
            {
                m_state = System.Threading.ThreadState.Stopped;
            }
            else
            {
                m_state = System.Threading.ThreadState.Aborted;
            }
            m_readyEvent.Reset();
        }

        /// <summary>
        /// Does a single synchronization pass for given items from the left side.
        /// </summary>
        /// <param name="items">Items to synchronize</param>
        private void ItemsSyncPass(
            IEnumerable<IMigrationWorkItem> items)
        {
            SynchronizeMetadata();
            m_readingEvent.Set();

            if (m_syncObjects == SyncObjects.Artifacts)
            {
                foreach (IMigrationWorkItem item in items)
                {
                    if ((item.Flags & MigrationWorkItemData.Attachments) != 0 || (item.Flags & MigrationWorkItemData.Links) != 0)
                    {
                        m_slowQueue.Add(item);
                    }
                    else
                    {
                        m_fastQueue.Add(item);
                    }
                }
            }
            else
            {
                LinkEngine.EnumerateDeferredLinkSources(
                    new EnumerateSourcesCallback(ProcessLinkSource));
            }

            // Flush pending items.
            Flush();
        }

        /// <summary>
        /// Does a single synchronization pass on deferred links.
        /// </summary>
        private void LinksSyncPass()
        {
            m_readingEvent.Set();
            LinkEngine.EnumerateDeferredLinkSources(
                new EnumerateSourcesCallback(ProcessLinkSource));

            // Flush pending items
            Flush();
        }

        /// <summary>
        /// Flushes pending queued items.
        /// </summary>
        private void Flush()
        {
            m_slowQueue.Flush();
            m_fastQueue.Flush();

            Wait(m_slowQueue.DoneEvent, m_fastQueue.DoneEvent);

            m_writeQueues.Left.Flush();
            m_writeQueues.Right.Flush();
            Wait(m_writeQueues.Left.DoneEvent, m_writeQueues.Right.DoneEvent);
        }

        /// <summary>
        /// Synchronizes metadata between two systems.
        /// </summary>
        private void SynchronizeMetadata()
        {
            TraceManager.EnterMethod();
            TfsMigrationWorkItemStore tfsStore = (TfsMigrationWorkItemStore)Stores[TfsSide];
            IMigrationWorkItemStore otherStore = (IMigrationWorkItemStore)Stores[!TfsSide];

            otherStore.SynchronizeMetadata(tfsStore.WorkItemStore.Projects[Session.TfsMigrationSource.Project]);
            tfsStore.WorkItemStore.SyncToCache();
            TraceManager.ExitMethod();
        }

        /// <summary>
        /// Swaps sides used in synchronization.
        /// </summary>
        private void SwapSides()
        {
            // Write queues must not be inversed - they rely on information from stores, which we update.
            lock (m_storesLock)
            {
                if (m_stores != null)
                {
                    m_stores = !m_stores;
                }
                if (m_storePools != null)
                {
                    m_storePools = !m_storePools;
                }
            }
            m_storeTypes = !m_storeTypes;
            m_tfsSide = !m_tfsSide;
        }

        /// <summary>
        /// Callback function for analysis queues.
        /// </summary>
        /// <param name="items">Items in the queue</param>
        /// <param name="queue">Source queue</param>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes")]
        private void AnalysisQueueCallback(
            List<IMigrationWorkItem> items,
            ReadQueue queue)
        {
            try
            {
                // Find items with reflections
                RevisionMapper[] mappers = new RevisionMapper[items.Count];

                for (int i = 0; i < items.Count; i++)
                {
                    IMigrationWorkItem item = items[i];

                    Session.FireMigrationStart(this, new FullId(Stores.Left, item.Watermark.Id));
                    mappers[i] = new RevisionMapper(this, item);
                }

                if (m_state == System.Threading.ThreadState.AbortRequested)
                {
                    return;
                }

                // Load reflections
                List<string> rightIds = new List<string>(items.Count);
                for (int i = 0; i < items.Count; i++)
                {
                    string rightId = mappers[i].ItemIds.Right;
                    if (!string.IsNullOrEmpty(rightId))
                    {
                        rightIds.Add(rightId);
                    }
                }

                GetWorkItemResult[] results;
                IMigrationWorkItemStore rightStore = StorePools.Right.GetStore();

                Dictionary<string, IMigrationWorkItem> rightItems = new Dictionary<string, IMigrationWorkItem>(
                    rightStore.IdComparer);
                Dictionary<string, Exception> failedItems = new Dictionary<string, Exception>(
                    rightStore.IdComparer);

                try
                {
                    if (rightIds.Count > 0)
                    {
                        try
                        {
                            results = rightStore.GetItems(rightIds.ToArray());
                        }
                        catch (Exception e)
                        {
                            // One reason for all items
                            results = new GetWorkItemResult[rightIds.Count];
                            for (int i = 0; i < rightIds.Count; i++)
                            {
                                results[i] = new GetWorkItemResult(rightIds[i], e);
                            }
                        }
                    }
                    else
                    {
                        results = new GetWorkItemResult[0];
                    }
                }
                finally
                {
                    StorePools.Right.ReturnStore(rightStore);
                }

                // Process results
                for (int i = 0; i < results.Length; i++)
                {
                    GetWorkItemResult res = results[i];

                    if (res.Exception != null)
                    {
                        failedItems.Add(res.Id, res.Exception);
                    }
                    else
                    {
                        rightItems.Add(res.Id, res.WorkItem);
                    }
                }

                // Build and submit deltas
                for (int i = 0; i < items.Count; i++)
                {
                    m_unlockEvent.WaitOne();
                    if (m_state == System.Threading.ThreadState.AbortRequested)
                    {
                        return;
                    }

                    IMigrationWorkItem leftItem = items[i];
                    IMigrationWorkItem rightItem = null;
                    RevisionMapper mapper = mappers[i];
                    string rightId = mapper.ItemIds.Right;

                    if (!string.IsNullOrEmpty(rightId))
                    {
                        Exception e;

                        if (failedItems.TryGetValue(rightId, out e))
                        {
                            // We failed to obtain item on the other side.
                            ReportException(leftItem.Watermark.Id, e);
                            continue;
                        }
                        else if (!rightItems.TryGetValue(rightId, out rightItem))
                        {
                            // The item on the other side no longer exist
                            OrphanedWorkItemException oe = new OrphanedWorkItemException(
                                new FullId(Stores.Right, rightId));
                            ReportException(leftItem.Watermark.Id, oe);
                            continue;
                        }
                    }

                    SyncWorkItems(new Pair<IMigrationWorkItem>(leftItem, rightItem), mapper);
                }
                items.Clear();
            }
            catch (Exception e)
            {
                Session.FireSessionError(e);
                throw;
            }
        }

        /// <summary>
        /// Synchronizes a pair of items by calculating deltas and submitting them into the write queues.
        /// </summary>
        /// <param name="items">Items to synchronize</param>
        /// <param name="mapper">Revision mapper</param>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes")]
        private void SyncWorkItems(
            Pair<IMigrationWorkItem> items,
            RevisionMapper mapper)
        {
            Debug.Assert(items.Left != null, "Source item can never be null!");
            InternalChangeList leftChanges = null;
            InternalChangeList rightChanges = null;

            AnalysisEngine engine = new AnalysisEngine(this);

            try
            {
                if (items.Right == null)
                {
                    leftChanges = null;
                    rightChanges = engine.CopyItem(items.Left, mapper);
                }
                else
                {
                    Pair<InternalChangeList> changes = engine.CompareItems(items, mapper);
                    leftChanges = changes.Left;
                    rightChanges = changes.Right;
                }
            }
            catch (Exception e)
            {
                ReportException(items.Left.Watermark.Id, e);
                return;
            }
            Debug.Assert(rightChanges != null, "Change list for the right side cannot be null!");

            if (leftChanges != null && !leftChanges.IsEmpty)
            {
                m_writeQueues.Left.Add(leftChanges);
                Session.FireAnalysisComplete(this, leftChanges.SourceId, leftChanges.TargetId, leftChanges.SourceRevisions, 
                    leftChanges.TargetRevisions);
            }
            if (!rightChanges.IsEmpty)
            {
                Session.FireAnalysisComplete(this, rightChanges.SourceId, rightChanges.TargetId, rightChanges.SourceRevisions,
                    rightChanges.TargetRevisions);
                m_writeQueues.Right.Add(rightChanges);
            }
            else if (leftChanges == null || leftChanges.IsEmpty)
            {
                FullId sourceId = new FullId(Stores.Left, items.Left.Watermark.Id);
                FullId targetId = new FullId(Stores.Right, items.Right.Watermark.Id);

                Session.FireMigrationComplete(this, sourceId, targetId, 0, 0);
            }
        }

        /// <summary>
        /// Reports an exception occurred when synchronizing given item from the left side.
        /// </summary>
        /// <param name="sourceId">Id of the item on the left side (source)</param>
        /// <param name="e">Exception that must be reported</param>
        private void ReportException(
            string sourceId,
            Exception e)
        {
            Session.FireItemMigrationFailed(this, new FullId(Stores.Left, sourceId), e);
        }

        /// <summary>
        /// Callback function for the left write queue.
        /// </summary>
        /// <param name="items">Items in the queue</param>
        /// <param name="queue">Source queue</param>
        private void LeftWriteQueueCallback(
            List<InternalChangeList> items,
            WriteQueue queue)
        {
            WriteQueueCallback(Side.Left, items, queue);
        }

        /// <summary>
        /// Callback function for the right write queue.
        /// </summary>
        /// <param name="items">Items in the queue</param>
        /// <param name="queue">Source queue</param>
        private void RightWriteQueueCallback(
            List<InternalChangeList> items,
            WriteQueue queue)
        {
            WriteQueueCallback(Side.Right, items, queue);
        }

        /// <summary>
        /// Callback function for write queues.
        /// </summary>
        /// <param name="side">Side that should be updated</param>
        /// <param name="items">Queued items</param>
        /// <param name="queue">Source queue</param>
        private void WriteQueueCallback(
            Side side,
            List<InternalChangeList> items,
            WriteQueue queue)
        {
            Debug.Assert(items.Count > 0, "Processing zero items!");
            IMigrationWorkItemStore store = StorePools[side].GetStore();

            m_writingEvent.Set();
            try
            {
                while (items.Count > 0)
                {
                    m_unlockEvent.WaitOne();
                    if (m_state == System.Threading.ThreadState.AbortRequested)
                    {
                        return;
                    }

                    // Try getting extra items from the queue
                    queue.AppendPendingItems(items);

                    // Shave off the top update from every change list.
                    IWorkItemUpdate[] updates = new IWorkItemUpdate[items.Count];
                    for (int i = 0; i < items.Count; i++)
                    {
                        updates[i] = items[i].GetNextUpdate();
                    }

                    // Submit the updates
                    TraceManager.TraceInformation(
                        "Submitting {0} updates into '{1}!{2}'", updates.Length, store.SystemName, store.StoreName);
                    UpdateResult[] results = store.Update(updates);

                    if (results.Length != updates.Length)
                    {
                        string msg = string.Format(
                            Resources.Culture,
                            Resources.ErrorWrongNumberOfUpdateResults,
                            store.SystemName,
                            store.StoreName,
                            items.Count,
                            updates.Length);
                        throw new SynchronizationEngineException(msg);
                    }

                    for (int iRes = 0, iChange = 0; iRes < results.Length && iChange < items.Count; iRes++)
                    {
                        m_unlockEvent.WaitOne();
                        InternalChangeList change = items[iChange];
                        UpdateResult res = results[iRes];
                        IWorkItemUpdate update = updates[iRes];

                        if (res.Exception != null)
                        {
                            ReportException(change.SourceId.Id, res.Exception);
                        }
                        else
                        {
                            Stats.IncrementCreatedRevisions();
                            // Update watermark on pending update statements
                            change.UpdateWatermark(update.SourceRevision, res.Watermark);
                        }

                        // Remove empty/invalid change lists
                        if (res.Exception != null || change.IsEmpty)
                        {
                            items.RemoveAt(iChange);

                            if (res.Exception == null)
                            {
                                Session.FireMigrationComplete(
                                    this,
                                    change.SourceId,
                                    change.TargetId,
                                    change.SourceRevisions,
                                    change.TargetRevisions);
                            }
                        }
                        else
                        {
                            iChange++;
                        }
                    }
                }
            }
            catch (Exception e)
            {
                Session.FireSessionError(e);
                throw;
            }
            finally
            {
                StorePools[side].ReturnStore(store);
            }
        }

        /// <summary>
        /// Waits until all given events are set.
        /// </summary>
        /// <param name="events">Events to wait for</param>
        private static void Wait(
            params WaitHandle[] events)
        {
            // WaitAll doesn't work from STA threads.
            for (int i = 0; i < events.Length; i++)
            {
                events[i].WaitOne();
            }
        }

        /// <summary>
        /// Disposes the object.
        /// </summary>
        public void Dispose()
        {
            // Close connections.
            if (m_storePools != null)
            {
                m_storePools.Left.Close();
                m_storePools.Right.Close();
            }

            // Close shared connections
            if (m_stores != null)
            {
                m_stores.Left.Close();
                m_stores.Right.Close();
            }

            // Dispose events
            ((IDisposable)m_readingEvent).Dispose();
            ((IDisposable)m_writingEvent).Dispose();
            ((IDisposable)m_readyEvent).Dispose();
            ((IDisposable)m_unlockEvent).Dispose();
            ((IDisposable)m_abortEvent).Dispose();
            ((IDisposable)m_completeEvent).Dispose();
        }

        /// <summary>
        /// Processes deferred link sources.
        /// </summary>
        /// <param name="type">Link type</param>
        /// <param name="sourceUri">Source item's URI</param>
        private void ProcessLinkSource(
            SystemType type,
            string sourceUri)
        {
            // Was the source migrated?
            string newSourceUri = LinkEngine.FindReflection(type, sourceUri);

            if (string.IsNullOrEmpty(newSourceUri))
            {
                // The source is missing on the opposite side.
                return;
            }

            string id1, id2;
            if (!LinkEngine.TranslateUri(sourceUri, out id1) || !LinkEngine.TranslateUri(newSourceUri, out id2))
            {
                return;
            }

            // Get deferred links
            ReadOnlyCollection<ILink> deferredLinks = LinkEngine.GetDeferredLinks(type, sourceUri);

            // Get links on the source item
            Side sourceSide = type == SystemType.Tfs? TfsSide : !TfsSide;
            Side targetSide = !sourceSide;
            IMigrationWorkItem source = GetWorkItem(sourceSide, id1);
            IMigrationWorkItem target = null;

            if (source == null)
            {
                return;
            }
            List<ILink> sourceLinks = new List<ILink>(LinkEngine.GetLinks(source));
            List<ILink> targetLinks = null;
            IWorkItemUpdatePackage change = null;
            sourceLinks.Sort(LinkComparer.Current);

            for (int i = 0; i < deferredLinks.Count; i++)
            {
                ILink link = deferredLinks[i];

                // Does the link still exist on the source?
                if (sourceLinks.BinarySearch(link, LinkComparer.Current) >= 0)
                {
                    // Has link's target been migrated?
                    string newTargetUri = LinkEngine.FindReflection(type, link.TargetUri);
                    if (string.IsNullOrEmpty(newTargetUri))
                    {
                        continue;
                    }

                    // Does the link exist on the target?
                    ILink targetLink = link.Redirect(newTargetUri);

                    if (target == null)
                    {
                        Debug.Assert(targetLinks == null);
                        target = GetWorkItem(targetSide, id2);
                        if (target == null)
                        {
                            continue;
                        }
                        targetLinks = new List<ILink>(LinkEngine.GetLinks(target));
                        targetLinks.Sort(LinkComparer.Current);
                    }

                    if (targetLinks.BinarySearch(targetLink, LinkComparer.Current) < 0)
                    {
                        if (change == null)
                        {
                            change = Stores[targetSide].ChangeListFactory.CreateUpdatePackage(source, target);
                        }
                        change.LinkOperations.Add(
                            new LinkOperation(AssociationAction.Add, targetLink));
                        continue;
                    }
                }

                // The link is obsolete; delete it
                LinkEngine.RemoveDeferred(type, link);
            }

            // Submit for processing
            if (change != null)
            {
                Pair<FullId> ids = new Pair<FullId>(
                    new FullId(Stores[sourceSide], source.Watermark.Id),
                    new FullId(Stores[targetSide], target.Watermark.Id));
                RevisionMapper mapper = new RevisionMapper(this, source);
                m_writeQueues[targetSide].Add(
                    new InternalChangeList(targetSide, ids, change, mapper));
            }
        }

        /// <summary>
        /// Gets given work item; fires an event in case of error.
        /// </summary>
        /// <param name="sourceSide">Side of the work item</param>
        /// <param name="id">Id of the work item</param>
        /// <returns>Work item object</returns>
        private IMigrationWorkItem GetWorkItem(
            Side sourceSide,
            string id)
        {
            GetWorkItemResult res = Stores[sourceSide].GetItems(new string[] { id })[0];
            if (res.Exception != null)
            {
                //$TODO: fire an event
                return null;
            }
            else
            {
                return res.WorkItem;
            }

        }
        #endregion
    }
}
